package com.sui.bigdata.flink.sql.side;

import com.sui.bigdata.flink.sql.core.Main;
import org.apache.flink.calcite.shaded.com.google.common.base.Charsets;
import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
import org.junit.Test;
import java.net.URLEncoder;
import java.util.List;

/**
 * @author YongChen
 * @date 2019/9/16 9:53
 * @description
 * @email yong_chen@sui.com
 */
public class P1 {


    @Test
    public void test1() throws Exception {
        String sql =
                " create scalar function RiskDecrypt with com.sui.bigdata.flink.sql.function.scalar.RiskDecrypt;\n" +
                        "create scalar function FinAesDecode with com.sui.bigdata.flink.sql.function.scalar.FinAesDecode;\n" +
                        "create scalar function LoanCclDecrypt with com.sui.bigdata.flink.sql.function.scalar.LoanCclDecrypt;\n" +
                        "create scalar function NotBlank with com.sui.bigdata.flink.sql.function.scalar.NotBlank;\n" +
                        "\n" +
                        "--- 维度表\n" +
                        "CREATE TABLE risk_ctrl_dim_geohash (\n" +
                        "   cf:geohash varchar as geohash,\n" +
                        "   cf:create_dt varchar as create_dt,\n" +
                        "   PRIMARY KEY(user_id),\n" +
                        "   PERIOD FOR SYSTEM_TIME\n" +
                        ") WITH (\n" +
                        "   type ='hbase',\n" +
                        "   zookeeperQuorum ='10.203.65.13:2181,10.203.65.14:2181,10.203.65.15:2181',\n" +
                        "   zookeeperParent ='/hbase',\n" +
                        "   tableName ='risk_ctrl_dim_geohash',\n" +
                        "   cache ='LRU',\n" +
                        "   cacheSize ='10000',\n" +
                        "   cacheTTLMs ='60000',\n" +
                        "   parallelism ='3',\n" +
                        "   partitionedJoin='false'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE risk_ctrl_dim_wifi (\n" +
                        "   cf:wifi varchar as wifi,\n" +
                        "   cf:create_dt varchar as create_dt,\n" +
                        "   PRIMARY KEY(user_id),\n" +
                        "   PERIOD FOR SYSTEM_TIME\n" +
                        ") WITH (\n" +
                        "   type ='hbase',\n" +
                        "   zookeeperQuorum ='10.203.65.13:2181,10.203.65.14:2181,10.203.65.15:2181',\n" +
                        "   zookeeperParent ='/hbase',\n" +
                        "   tableName ='risk_ctrl_dim_wifi',\n" +
                        "   cache ='LRU',\n" +
                        "   cacheSize ='10000',\n" +
                        "   cacheTTLMs ='60000',\n" +
                        "   parallelism ='3',\n" +
                        "   partitionedJoin='false'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE feidee_fid_fname (\n" +
                        "   cf:fname varchar as fname,\n" +
                        "   PRIMARY KEY(fid),\n" +
                        "   PERIOD FOR SYSTEM_TIME\n" +
                        ") WITH (\n" +
                        "   type ='hbase',\n" +
                        "   zookeeperQuorum ='10.203.65.13:2181,10.203.65.14:2181,10.203.65.15:2181',\n" +
                        "   zookeeperParent ='/hbase',\n" +
                        "   tableName ='feidee_fid_fname',\n" +
                        "   cache ='LRU',\n" +
                        "   cacheSize ='10000',\n" +
                        "   cacheTTLMs ='60000',\n" +
                        "   parallelism ='3',\n" +
                        "   partitionedJoin='false'\n" +
                        ");\n" +
                        "\n" +
                        "--- 事实表\n" +
                        "--- 福贷\n" +
                        "CREATE TABLE bdl_suigo_business_apply (\n" +
                        "   rowData.customer_id varchar as customer_id,\n" +
                        "   rowData.business_type varchar as business_type,\n" +
                        "   rowData.update_date timestamp as applytime,\n" +
                        "   WATERMARK FOR applytime AS withOffset(applytime, 0)\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.200.1.52:9092,10.200.1.53:9092,10.200.1.54:9092,10.200.1.55:9092,10.200.1.56:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-graph-202001060000',\n" +
                        "   topic ='rtsync.bdl_suigo_business_apply',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE bdl_suigo_loan_customer_app_active (\n" +
                        "   rowData.customer_id varchar as customer_id,\n" +
                        "   rowData.client_uid varchar as client_uid,\n" +
                        "   rowData.bound_mobile varchar as ph,\n" +
                        "   rowData.information_certid_no varchar as card,\n" +
                        "   rowData.information_contact_phone varchar as ct_ph,\n" +
                        "   rowData.create_date varchar as create_date,\n" +
                        "   rowData.update_date timestamp as activetime,\n" +
                        "   WATERMARK FOR activetime AS withOffset(activetime, 0)\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.200.1.52:9092,10.200.1.53:9092,10.200.1.54:9092,10.200.1.55:9092,10.200.1.56:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-graph-202001060000',\n" +
                        "   topic ='rtsync.bdl_suigo_loan_customer_app_active',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE bdl_suigo_loan_customer_work_info (\n" +
                        "   rowData.customer_id varchar as customer_id,\n" +
                        "   rowData.workspace_name varchar as workspace_name,\n" +
                        "   rowData.workspace_phone varchar as workspace_phone,\n" +
                        "   rowData.update_date timestamp as worktime,\n" +
                        "   WATERMARK FOR worktime AS withOffset(worktime, 0)\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.200.1.52:9092,10.200.1.53:9092,10.200.1.54:9092,10.200.1.55:9092,10.200.1.56:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-graph-202001060000',\n" +
                        "   topic ='rtsync.bdl_suigo_loan_customer_work_info',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "--- 瑞贷\n" +
                        "CREATE TABLE bdl_riskctrl_active_apply_info  (\n" +
                        "   rowData.user_id varchar as user_id,\n" +
                        "   rowData.id_card varchar as id_card,\n" +
                        "   rowData.phone varchar as phone,\n" +
                        "   rowData.create_time varchar as create_time,\n" +
                        "   rowData.update_time timestamp as applytime,\n" +
                        "   WATERMARK FOR applytime AS withOffset(applytime, 0)\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.200.1.52:9092,10.200.1.53:9092,10.200.1.54:9092,10.200.1.55:9092,10.200.1.56:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-graph-202001060000',\n" +
                        "   topic ='rtsync.bdl_riskctrl_active_apply_info',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE bdl_loan_credit_center_user_extra_info  (\n" +
                        "   rowData.user_id varchar as user_id,\n" +
                        "   rowData.company_name varchar as company_name,\n" +
                        "   rowData.company_phone varchar as company_phone,\n" +
                        "   rowData.create_time varchar as create_time\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.200.1.52:9092,10.200.1.53:9092,10.200.1.54:9092,10.200.1.55:9092,10.200.1.56:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-graph-202001060000',\n" +
                        "   topic ='rtsync.bdl_loan_credit_center_user_extra_info',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE bdl_loan_credit_center_user_emergy_contact  (\n" +
                        "   rowData.user_id varchar as user_id,\n" +
                        "   rowData.phone varchar as phone,\n" +
                        "   rowData.create_time varchar as create_time\n" +
                        ") WITH (\n" +
                        "   type ='kafka',\n" +
                        "   bootstrapServers ='10.200.1.52:9092,10.200.1.53:9092,10.200.1.54:9092,10.200.1.55:9092,10.200.1.56:9092',\n" +
                        "   offsetReset ='latest',\n" +
                        "   groupid='datateam-graph-202001060000',\n" +
                        "   topic ='rtsync.bdl_loan_credit_center_user_emergy_contact',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "--- 结果表\n" +
                        "CREATE TABLE graph_risk_ctrl_edge_console (\n" +
                        "   fromv varchar,\n" +
                        "   tov varchar,\n" +
                        "   edge_type varchar,\n" +
                        "   create_dt varchar\n" +
                        ") WITH (\n" +
                        "   type='console',\n" +
                        "   parallelism ='1'\n" +
                        ");\n" +
                        "\n" +
                        "CREATE TABLE graph_risk_ctrl_edge_kafka (\n" +
                        "   fromv varchar,\n" +
                        "   tov varchar,\n" +
                        "   edge_type varchar,\n" +
                        "   create_dt varchar\n" +
                        ") WITH (\n" +
                        "   type='kafka',\n" +
                        "   bootstrapServers='10.200.8.17:9092',\n" +
                        "   topic='graph_risk_control_edge'\n" +
                        ");\n" +
                        "\n" +
                        "--- 口径\n" +
                        "\n" +
                        "\n" +
                        "create view fd_act as\n" +
                        "select \n" +
                        "    person_info.client_uid as client_uid,\n" +
                        "    person_info.ph as ph,\n" +
                        "    person_info.card as card,\n" +
                        "    person_info.ct_ph as ct_ph,\n" +
                        "    person_info.create_dt as create_dt,\n" +
                        "    work_info.workspace_name as workspace_name,\n" +
                        "    regexp_replace(work_info.workspace_phone,'-','') as workspace_phone\n" +
                        "from (\n" +
                        "    select \n" +
                        "        active.customer_id,\n" +
                        "        active.client_uid,\n" +
                        "        active.ph,\n" +
                        "        active.card,\n" +
                        "        active.ct_ph,\n" +
                        "        active.create_dt,\n" +
                        "        active.activetime\n" +
                        "    from (\n" +
                        "        select customer_id,business_type,applytime\n" +
                        "        from bdl_suigo_business_apply\n" +
                        "        where business_type = '4'\n" +
                        "    ) apply\n" +
                        "    inner join (\n" +
                        "        select\n" +
                        "            customer_id,\n" +
                        "            client_uid,\n" +
                        "            FinAesDecode(ph,'ed5e62c607c7c2e441817518ba9ad16bc635aaf93146c5fd3cf8985bf858f0ce') as ph,\n" +
                        "            FinAesDecode(card,'ed5e62c607c7c2e441817518ba9ad16bc635aaf93146c5fd3cf8985bf858f0ce') as card,\n" +
                        "            FinAesDecode(ct_ph,'ed5e62c607c7c2e441817518ba9ad16bc635aaf93146c5fd3cf8985bf858f0ce') as ct_ph,\n" +
                        "            substring(create_date,1,10) as create_dt,\n" +
                        "            activetime\n" +
                        "        from bdl_suigo_loan_customer_app_active\n" +
                        "    ) active\n" +
                        "    on apply.customer_id = active.customer_id and active.activetime between apply.applytime and apply.applytime + interval '10' second\n" +
                        ") person_info\n" +
                        "left join bdl_suigo_loan_customer_work_info work_info\n" +
                        "on person_info.customer_id = work_info.customer_id and work_info.worktime between person_info.activetime and person_info.activetime + interval '10' second;\n" +
                        "\n" +
                        "create view rd_act as\n" +
                        "select \n" +
                        "    user_id,\n" +
                        "    RiskDecrypt(id_card) as card,\n" +
                        "    RiskDecrypt(phone) as ph,\n" +
                        "    substring(create_time,1,10) as create_dt\n" +
                        "from bdl_riskctrl_active_apply_info \n" +
                        "where NotBlank(user_id);\n" +
                        "\n" +
                        "create view rd_exuser as\n" +
                        "select user_id,company_name,company_phone,substring(create_time,1,10) as create_dt\n" +
                        "from bdl_loan_credit_center_user_extra_info \n" +
                        "where NotBlank(user_id);\n" +
                        "\n" +
                        "create view rd_contact as\n" +
                        "select \n" +
                        "    user_id,\n" +
                        "    LoanCclDecrypt(phone) as phone,\n" +
                        "    substring(create_time,1,10) as create_dt\n" +
                        "from bdl_loan_credit_center_user_emergy_contact\n" +
                        "where NotBlank(user_id);\n" +
                        "\n" +
                        "create view fd_geohash as\n" +
                        "select f.client_uid as user_id,geo.geohash as tov,'is_geohash' as edge_type,geo.create_dt as create_dt \n" +
                        "from fd_act f join risk_ctrl_dim_geohash geo on f.client_uid = geo.user_id;\n" +
                        "\n" +
                        "create view fd_wifi as\n" +
                        "select f.client_uid as user_id,w.wifi as tov,'is_wifi' as edge_type,w.create_dt as create_dt \n" +
                        "from fd_act f join risk_ctrl_dim_wifi w on f.client_uid = w.user_id;\n" +
                        "\n" +
                        "create view rd_geohash as\n" +
                        "select r.user_id as user_id,geo.geohash as tov,'is_geohash' as edge_type,geo.create_dt as create_dt \n" +
                        "from rd_act r join risk_ctrl_dim_geohash geo on r.user_id = geo.user_id;\n" +
                        "\n" +
                        "create view rd_wifi as\n" +
                        "select r.user_id as user_id,w.wifi as tov,'is_wifi' as edge_type,w.create_dt as create_dt \n" +
                        "from rd_act r join risk_ctrl_dim_wifi w on r.user_id = w.user_id;\n" +
                        "\n" +
                        "insert into graph_risk_ctrl_edge_kafka\n" +
                        "select distinct id_name.fname as fromv,fd_rd.tov as tov,fd_rd.edge_type as edge_type,fd_rd.create_dt as create_dt\n" +
                        "from (\n" +
                        "    --- 福贷\n" +
                        "    select client_uid as user_id,ph as tov,'is_ph' as edge_type,create_dt from fd_act where NotBlank(ph) \n" +
                        "    and substring(ph,1,3)<>'400' \n" +
                        "    and substring(ph,4,CHAR_LENGTH(ph)-3) not in ('0000000','00000000','000000000') \n" +
                        "    and substring(ph,5,CHAR_LENGTH(ph)-4) not in ('000000','0000000','00000000')\n" +
                        "    union all\n" +
                        "    select client_uid as user_id,card as tov,'is_cd' as edge_type,create_dt from fd_act where NotBlank(card)\n" +
                        "    union all\n" +
                        "    select client_uid as user_id,ct_ph as tov,'is_ct_ph' as edge_type,create_dt from fd_act where NotBlank(ct_ph) \n" +
                        "    and substring(ct_ph,1,3)<>'400' \n" +
                        "    and substring(ct_ph,4,CHAR_LENGTH(ct_ph)-3) not in ('0000000','00000000','000000000') \n" +
                        "    and substring(ct_ph,5,CHAR_LENGTH(ct_ph)-4) not in ('000000','0000000','00000000')\n" +
                        "    union all\n" +
                        "    select client_uid as user_id,workspace_name as tov,'is_cp' as edge_type,create_dt from fd_act \n" +
                        "    where NotBlank(workspace_name)\n" +
                        "        and CHAR_LENGTH(trim(workspace_name))>=4 and CHAR_LENGTH(trim(workspace_name))<=20\n" +
                        "        and trim(workspace_name) not like '%速递%'\n" +
                        "        and trim(workspace_name) not like '%快递%'\n" +
                        "        and trim(workspace_name) not like '%配送%'\n" +
                        "        and trim(workspace_name) not like '%速运%'\n" +
                        "        and trim(workspace_name) not like '%物流%'\n" +
                        "        and trim(workspace_name) not like '%保险%'\n" +
                        "        and trim(workspace_name) not like '%银行%'\n" +
                        "        and trim(workspace_name) not like '%中国%'\n" +
                        "        and trim(workspace_name) not like '%富士康%'\n" +
                        "        and trim(workspace_name) not like '%华为%'\n" +
                        "        and trim(workspace_name) not like '%个体%'\n" +
                        "        and trim(workspace_name) not like '%人寿%'\n" +
                        "        and trim(workspace_name) not like '%自由职业%'\n" +
                        "        and trim(workspace_name) not like '%外卖%'\n" +
                        "        and trim(workspace_name) not like '%地产%'\n" +
                        "        and trim(workspace_name) not like '%滴滴%'\n" +
                        "        and trim(workspace_name) not like '%美团%'\n" +
                        "    union all\n" +
                        "    select client_uid as user_id,workspace_phone as tov,'is_cp_ph' as edge_type,create_dt from fd_act where NotBlank(workspace_phone) \n" +
                        "    and substring(workspace_phone,1,3)<>'400' \n" +
                        "    and substring(workspace_phone,4,CHAR_LENGTH(workspace_phone)-3) not in ('0000000','00000000','000000000') \n" +
                        "    and substring(workspace_phone,5,CHAR_LENGTH(workspace_phone)-4) not in ('000000','0000000','00000000')\n" +
                        "    --- 瑞贷\n" +
                        "    union all\n" +
                        "    select user_id,ph as tov,'is_ph' as edge_type,create_dt from rd_act where NotBlank(ph) \n" +
                        "    and substring(ph,1,3)<>'400' \n" +
                        "    and substring(ph,4,CHAR_LENGTH(ph)-3) not in ('0000000','00000000','000000000') \n" +
                        "    and substring(ph,5,CHAR_LENGTH(ph)-4) not in ('000000','0000000','00000000')\n" +
                        "    union all\n" +
                        "    select user_id,card as tov,'is_cd' as edge_type,create_dt from rd_act where NotBlank(card)\n" +
                        "    union all\n" +
                        "    select user_id,company_name as tov,'is_cp' as edge_type,create_dt from rd_exuser \n" +
                        "    where NotBlank(company_name)\n" +
                        "        and CHAR_LENGTH(trim(company_name))>=4 and CHAR_LENGTH(trim(company_name))<=20\n" +
                        "            and trim(company_name) not like '%速递%'\n" +
                        "            and trim(company_name) not like '%快递%'\n" +
                        "            and trim(company_name) not like '%配送%'\n" +
                        "            and trim(company_name) not like '%速运%'\n" +
                        "            and trim(company_name) not like '%物流%'\n" +
                        "            and trim(company_name) not like '%保险%'\n" +
                        "            and trim(company_name) not like '%银行%'\n" +
                        "            and trim(company_name) not like '%中国%'\n" +
                        "            and trim(company_name) not like '%富士康%'\n" +
                        "            and trim(company_name) not like '%华为%'\n" +
                        "            and trim(company_name) not like '%个体%'\n" +
                        "            and trim(company_name) not like '%人寿%'\n" +
                        "            and trim(company_name) not like '%自由职业%'\n" +
                        "            and trim(company_name) not like '%外卖%'\n" +
                        "            and trim(company_name) not like '%地产%'\n" +
                        "            and trim(company_name) not like '%滴滴%'\n" +
                        "            and trim(company_name) not like '%美团%'\n" +
                        "    union all\n" +
                        "    select user_id,company_phone as tov,'is_cp_ph' as edge_type,create_dt from rd_exuser where NotBlank(company_phone) \n" +
                        "    and substring(company_phone,1,3)<>'400' \n" +
                        "    and substring(company_phone,4,CHAR_LENGTH(company_phone)-3) not in ('0000000','00000000','000000000') \n" +
                        "    and substring(company_phone,5,CHAR_LENGTH(company_phone)-4) not in ('000000','0000000','00000000')\n" +
                        "    union all\n" +
                        "    select user_id,phone as tov,'is_ct_ph' as edge_type,create_dt from rd_contact where NotBlank(phone) \n" +
                        "    and substring(phone,1,3)<>'400' \n" +
                        "    and substring(phone,4,CHAR_LENGTH(phone)-3) not in ('0000000','00000000','000000000') \n" +
                        "    and substring(phone,5,CHAR_LENGTH(phone)-4) not in ('000000','0000000','00000000')\n" +
                        ") fd_rd\n" +
                        "join feidee_fid_fname id_name on fd_rd.user_id = id_name.fid;\n" +
                        "\n" +
                        "--- geohash&wifi\n" +
                        "insert into graph_risk_ctrl_edge_kafka\n" +
                        "select distinct id_name.fname as fromv,g_w.tov as tov,g_w.edge_type as edge_type,g_w.create_dt as create_dt\n" +
                        "from (\n" +
                        "    select user_id,tov,edge_type,create_dt from fd_geohash\n" +
                        "    union all\n" +
                        "    select user_id,tov,edge_type,create_dt from fd_wifi\n" +
                        "    union all\n" +
                        "    select user_id,tov,edge_type,create_dt from rd_geohash\n" +
                        "    union all\n" +
                        "    select user_id,tov,edge_type,create_dt from rd_wifi\n" +
                        ") g_w\n" +
                        "join feidee_fid_fname id_name on g_w.user_id = id_name.fid;";

        test2(sql);
    }

    @Test
    public void test2() throws Exception {
        String sql =
                " create scalar function notBlank with com.sui.bigdata.flink.sql.function.scalar.NotBlank;\n" +
                        "create scalar function p1Time with com.sui.bigdata.flink.sql.function.scalar.P1Time;\n" +
                        "CREATE TABLE trick(  \n" +
                        "    rowData.user_id varchar as user_id \n" +
                        " )WITH( \n" +
                        "    type ='kafka', \n" +
                        "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093', \n" +
                        "    offsetReset ='latest',\n" +
                        "    groupid ='chenyong999999', \n" +
                        "    topic ='rtsync.bdl_riskctrl_loan_apply_info,rtsync.bdl_riskctrl_active_apply_info', \n" +
                        "    parallelism ='2' \n" +
                        " ); \n" +
                        "\n" +
                        " CREATE TABLE tidb_side ( \n" +
                        "  userid  varchar , \n" +
                        "  newaction varchar ,  \n" +
                        "  etype varchar ,  \n" +
                        "  cost_time varchar ,  \n" +
                        "  dt    varchar ,   \n" +
                        "  PRIMARY KEY(userid),   \n" +
                        "  PERIOD FOR SYSTEM_TIME   \n" +
                        "  ) WITH (  \n" +
                        "  type='mysql',  \n" +
                        "  url='jdbc:mysql://tidb1.testfeideedba.com:4000/dt?useUnicode=true&amp;characterEncoding=utf8;useSSL=true',\n" +
                        "  userName='DSJ_user',\n" +
                        "  password='606cde3507b026H,dHb_77e67H_e',\n" +
                        "  tableName='adl_bdl_riskctrl_sswl_credit_behaviour_etl_behaviour_tidb'); \n" +
                        "\n" +
                        "CREATE TABLE hbase_side (  \n" +
                        "   info:dt1 varchar as dt1, \n" +
                        "   info:dt2 varchar as dt2, \n" +
                        "   PRIMARY KEY (user_id_md5), \n" +
                        "   PERIOD FOR SYSTEM_TIME ) WITH  \n" +
                        "  (type = 'hbase', \n" +
                        "   zookeeperQuorum = '10.201.7.113:2181,10.201.7.114:2181,10.201.7.115:2181', \n" +
                        "   zookeeperParent = '/hbase', \n" +
                        "   tableName = 'adl_bd_riskctrl_sswl_credit_behaviour_etl_dt'); \n" +
                        "\n" +
                        "CREATE TABLE adl_bd_ls_ssj_kn_user_suid (  \n" +
                        "   info:suid varchar as suid , \n" +
                        "   PRIMARY KEY (user_id_md5), \n" +
                        "   PERIOD FOR SYSTEM_TIME ) WITH  \n" +
                        "  (type = 'hbase', \n" +
                        "   zookeeperQuorum = '10.201.7.113:2181,10.201.7.114:2181,10.201.7.115:2181', \n" +
                        "   zookeeperParent = '/hbase', \n" +
                        "   tableName = 'adl_bd_ls_ssj_kn_user_suid'); \n" +
                        "select   \n" +
                        "   m.suid,   \n" +
                        "    md5(m.suid) as suid_md5,   \n" +
                        "   (case when m.companyaddress_inputbox_counter=0 then 'null' when m.companyaddress_inputbox_ct<>0 and m.companyaddress_inputbox_ct=m.companyaddress_inputbox_counter  then '1'  \n" +
                        "  else '0' end) as userkn_companyaddress_paste_realtime , \n" +
                        "  (case when m.companynumber_inputbox_counter=0 then 'null' when m.companynumber_address=0 and companynumber_inputbox_ct<>0 and m.companynumber_inputbox_ct=m.companynumber_inputbox_counter  then '1'   else '0' end) as userkn_companynumber_paste_realtime , \n" +
                        "  (case when m.companyname_counter=0 then 'null' when m.companyname_ct<>0 and m.companyname_ct=m.companyname_counter  then '1'  else '0' end) as userkn_companyname_paste_realtime , \n" +
                        "  (case when m.creditcardno_inputbox_counter=0 then 'null' when m.bankcard_creditcardno_inputbox_ct<>0 and m.bankcard_creditcardno_inputbox_ct=m.creditcardno_inputbox_counter  then '1'  else '0' end) as userkn_creditcardno_paste_realtime , \n" +
                        "  (case when m.residenceaddress_inputbox_counter=0 then 'null' when m.residenceaddress_inputbox_ct<>0 and m.residenceaddress_inputbox_ct=m.residenceaddress_inputbox_counter  then '1'  else '0' end) as userkn_residenceaddress_paste_realtime, \n" +
                        "  (case when m.action_count=0 then 'null'  when m.companyaddress_GPS<>0 and m.companyaddress_choicebox<>0 then '1' else '0' end) as userkn_companyaddress_gps_and_edit_choicebox_realtime,   \n" +
                        "  (case when m.action_count=0 then 'null' when m.companyaddress_GPS<>0 and m.companyaddress_choicebox=0 and m.companyaddress_inputbox=0 then '1' else '0' end) as userkn_companyaddress_gps_and_no_edit_realtime,     \n" +
                        "  (case when m.action_count=0 then 'null' when m.residenceaddress_GPS<>0 and m.residenceaddress_choicebox<>0  then '1' else '0' end)    as userkn_residenceaddress_gps_and_edit_choicebox_realtime,   \n" +
                        "  (case when m.action_count=0 then 'null' when m.residenceaddress_GPS<>0 and m.residenceaddress_choicebox=0 and  m.residenceaddress_inputbox=0 then '1' else '0' end)  as userkn_residenceaddress_gps_and_no_edit_realtime \n" +
                        "from(  \n" +
                        "  select   \n" +
                        "    t.suid ,  \n" +
                        "    sum(case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyaddress_inputbox' and  t3.etype='edit' and  cast(t3.cost_time as FLOAT)<3 then 1 else 0 end)  as companyaddress_inputbox_ct,  \n" +
                        "    sum( case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyaddress_inputbox' and  t3.etype='edit' then 1 else 0 end ) as  companyaddress_inputbox_counter, \n" +
                        "    sum(case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companynumber_inputbox' and  t3.etype='edit' and  cast(t3.cost_time as FLOAT)<3 then 1 else 0 end)  as companynumber_inputbox_ct,  \n" +
                        "     sum( case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companynumber_inputbox' and  t3.etype='edit' then 1 else 0 end ) as  companynumber_inputbox_counter,  \n" +
                        "     sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companynumber_address' and  t3.etype='click'  then 1 else 0 end ) as companynumber_address , \n" +
                        "    sum(case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyname' and  t3.etype='edit' and  cast(t3.cost_time as FLOAT)<3 then 1 else 0 end)  as companyname_ct,  \n" +
                        "     sum( case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyname' and  t3.etype='edit' then 1 else 0 end ) as  companyname_counter, \n" +
                        "    sum(case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='bankcard_creditcardno_inputbox' and  t3.etype='edit' and  cast(t3.cost_time as FLOAT)<3 then 1 else 0 end)  as bankcard_creditcardno_inputbox_ct,  \n" +
                        "     sum( case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='bankcard_creditcardno_inputbox' and  t3.etype='edit' then 1 else 0 end ) as  creditcardno_inputbox_counter, \n" +
                        "    sum(case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_residenceaddress_inputbox' and  t3.etype='edit' and  cast(t3.cost_time as FLOAT)<3 then 1 else 0 end)  as residenceaddress_inputbox_ct,  \n" +
                        "     sum( case when notBlank(t3.cost_time) and p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_residenceaddress_inputbox' and  t3.etype='edit' then 1 else 0 end ) as  residenceaddress_inputbox_counter, \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyaddress_GPS' and  t3.etype='click' then 1 else 0 end )  as companyaddress_GPS,    \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyaddress_choicebox' and  t3.etype='click' then 1 else 0 end)  as companyaddress_choicebox,   \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_companyaddress_inputbox' and  t3.etype='edit' then 1 else 0 end)  as companyaddress_inputbox,   \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_residenceaddress_GPS' and  t3.etype='click' then 1 else 0 end)  as residenceaddress_GPS,   \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_residenceaddress_choicebox' and  t3.etype='click' then 1 else 0 end)  as residenceaddress_choicebox,   \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and t3.newaction='gerenxinxi_residenceaddress_inputbox' and  t3.etype='edit' then 1 else 0 end)  as residenceaddress_inputbox, \n" +
                        "    sum(case when p1Time(t3.dt,t.dt1,t.dt2) and notBlank(t3.newaction) then 1 else 0 end) as action_count  \n" +
                        "  from (select t1.proctime,t1.suid,t1.user_id,t1.user_id_md5,t2.dt1,t2.dt2    \n" +
                        "   from (select t0.proctime,t0.user_id_md5,t0.user_id,t5.suid  \n" +
                        "        from (select proctime,user_id,md5(user_id) as user_id_md5  \n" +
                        "        from trick) t0  \n" +
                        "        join adl_bd_ls_ssj_kn_user_suid t5  \n" +
                        "        on  t0.user_id_md5=t5.user_id_md5 ) t1     \n" +
                        "   left join hbase_side t2     \n" +
                        "   on t1.user_id_md5=t2.user_id_md5) t     \n" +
                        "   left join tidb_side t3     \n" +
                        "   on t.user_id=t3.userid  \n" +
                        "GROUP BY SESSION(t.proctime, INTERVAL '1' SECOND),t.suid) m;";
        test2(sql);
    }

    @Test
    public void test3() throws Exception {
        String sql =
                    " CREATE TABLE  fin_fd_scorecard_data(\n" +
                    "    fname VARCHAR AS suid,\n" +
                    "    outer_ip VARCHAR,\n" +
                    "    gps_province VARCHAR,\n" +
                    "    gps_city VARCHAR\n" +
                    ") WITH (\n" +
                    "    type = 'kafka',\n" +
                    "    bootstrapServers = '10.201.7.187:9093,10.201.7.188:9093',\n" +
                    "    groupId = 'lc.riskctrl.fin_fd_scorecard_data_ATFD-604',\n" +
                    "    topic = 'lc.riskctrl.fin_fd_scorecard_data',\n" +
                    "    parallelism = '1');\n" +
                    "\t\n" +
                    "CREATE TABLE adl_bd_fraud_ipgps2_fdt_middle (\n" +
                    "    info:hit VARCHAR AS hit,\n" +
                    "    PRIMARY KEY (rowkey),\n" +
                    "    PERIOD FOR SYSTEM_TIME\n" +
                    ") WITH (\n" +
                    "    type = 'hbase',\n" +
                    "    zookeeperQuorum = '10.201.7.114:2181,10.201.7.115:2181,10.201.7.113:2181',\n" +
                    "    zookeeperParent = '/hbase',\n" +
                    "    tableName = 'adl_bd_fraud_ipgps2_fdt_middle');\n" +
                    "CREATE TABLE loan_credit_wallet (\n" +
                    "    info:suid_md5 VARCHAR,\n" +
                    "    info:suid VARCHAR,\n" +
                    "    info:hit_risk_ip_gps_2_jud VARCHAR\n" +
                    ") WITH (\n" +
                    "    type = 'hbase',\n" +
                    "    zookeeperQuorum = '10.201.7.114:2181,10.201.7.115:2181,10.201.7.113:2181',\n" +
                    "    zookeeperParent = '/hbase',\n" +
                    "    tableName = 'adl_bd_ls_loan_credit_wallet',\n" +
                    "    rowkey = 'info:suid_md5'\n" +
                    "); " +
                    " insert into loan_credit_wallet select \n" +
                    "   concat(t3.suid_md5,'creditwallet') AS creditwalle_md5,\n" +
                    "   t3.suid,\n" +
                    "   (case when t3.hit_suid = '1' or t4.hit = '1' then '1' else '0' end) as hit_risk_ip_gps_2_jud\n" +
                    " from \n" +
                    "   (select t1.suid,t1.suid_md5,t1.ipgps_md5,t2.hit as hit_suid\n" +
                    "    from (select suid,md5(suid) as suid_md5,md5(concat(outer_ip,'#',gps_province,'#',gps_city)) as ipgps_md5\n" +
                    "         from fin_fd_scorecard_data) t1\n" +
                    "    left join adl_bd_fraud_ipgps2_fdt_middle t2 \n" +
                    "    on t1.suid_md5 = t2.rowkey) t3\n" +
                    " left join adl_bd_fraud_ipgps2_fdt_middle t4\n" +
                    " on t3.ipgps_md5 = t4.rowkey;";

        test2(sql);
    }

   @Test
    public  void testfun() throws Exception {
        System.out.println("2019-09-16 00:00:00".compareTo("2019-09-15 23:00:00")>0);
    }



    public void test2(String sql) throws Exception {
        List<String> paramList = Lists.newArrayList();
        paramList.add("-sql");
        String exeSql = URLEncoder.encode(sql, Charsets.UTF_8.name());
        paramList.add(exeSql);
        paramList.add("-name");
        paramList.add("xc");
        paramList.add("-mode");
        paramList.add("local");
        paramList.add("-port");
        paramList.add("8084");
        paramList.add("-confProp");
        String conf = "{\"time.characteristic\":\"ProcessingTime\",\"parallelism\":2}";
        String confEncode = URLEncoder.encode(conf, Charsets.UTF_8.name());
        paramList.add(confEncode);

        String[] params = new String[paramList.size()];
        paramList.toArray(params);
        Main.main(params);
    }
}
